package sparkml_study
import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object als {
  def main(args: Array[String]): Unit = {
    /*
        协同过滤（Collaborative Filtering, CF） 是一种常用于推荐系统中的算法，旨在根据用户和物品之间的历史互动或评分
        来进行推荐。协同过滤的核心思想是：如果用户A与用户B有相似的行为（例如评分、点击等），那么用户A未接触过的物品，用户
        B可能喜欢，反之亦然。

    协同过滤的算法主要可以分为两类：

    基于用户的协同过滤（User-Based Collaborative Filtering）
    基于物品的协同过滤（Item-Based Collaborative Filtering）
    此外，还有一种衍生的方式是 矩阵分解，如 ALS（交替最小二乘法），这类方法通常被认为是协同过滤的扩展。


    协同过滤 是推荐系统中的核心方法，通过用户或物品之间的相似性来进行推荐。
    基于用户的协同过滤 关注用户之间的相似性。
    基于物品的协同过滤 关注物品之间的相似性。
    矩阵分解方法（如ALS） 是一种高效的技术，尤其在大规模数据下，通过分解评分矩阵来进行推荐。
     */


    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("协同过滤演示")
      .getOrCreate()

    //  导入隐式转换，以防多余的报错
    import spark.implicits._

    //  定义将读取出来的数据转化为Rating对象的方法
    def parseRating(str:String):Rating={
      val fields=str.split("::")
      assert(fields.size==4)
      Rating(fields(0).toInt,fields(1).toInt,fields(2).toFloat,fields(3).toLong)
    }

    //  读取spark自带的测试数据并将其转化为dataframe
    val ratings=spark.sparkContext
      .textFile("hdfs://192.168.40.110:9000/spark_test_data/mllib/als/sample_movielens_ratings.txt")
      .map(parseRating).toDF()

    ratings.show

    //  将数据分成两部分，训练集占百分之80，测试集占百分之20
    val Array(training,test)=ratings.randomSplit(Array(0.8,0.2))

    //  使用ALS来建立推荐模型，这里构建两个模型，一个是显性反馈，另一个是隐性反馈
    //  显性
    val alsExplicit=new ALS()
      .setMaxIter(5)
      .setRegParam(0.01)
      .setUserCol("useid")
      .setItemCol("movieid")
      .setRatingCol("rating")

    //  隐性
    val alsimplicit=new ALS()
      .setMaxIter(5)
      .setRegParam(0.01)
      .setImplicitPrefs(true)
      .setUserCol("useid")
      .setItemCol("movieid")
      .setRatingCol("rating")

    //  把推荐模型放在训练数据上训练
    //  显性
    val modelExplicit=alsExplicit.fit(training)
    //  隐性
    val modelImplicit=alsimplicit.fit(training)

    //  对测试集中的用户-电影进行预测，得到预测评分的数据集
    val predictionsExplicit=modelExplicit
      .transform(test)
      .na.drop()

    val predictionsImplicit=modelImplicit
      .transform(test)
      .na.drop()

    //  对预测结果进行查看(rating是真实的打分,prediction是预测的打分)
    predictionsExplicit.show


    //  通过计算模型的均方根误差，来对模型进行评估，均方根误差越小，模型越准确
    //  设置评估器
    val evaluator=new RegressionEvaluator()
      .setMetricName("rmse")
      .setLabelCol("rating")
      .setPredictionCol("prediction")

    //  对模型进行评估
    val rmseExplicit=evaluator.evaluate(predictionsExplicit)

    val rmseImplicit=evaluator.evaluate(predictionsImplicit)

    //  打印查看评估结果
    println(rmseExplicit)
    println(rmseImplicit)


    println("----------------------------协同过滤完成推荐----------------------------")
    //  使用训练好的模型，对每个用户推荐top10的电影recommendations:建议
    val user_recommendations:DataFrame=modelExplicit.recommendForAllUsers(10)
    //  显示推荐的结果
    user_recommendations.show(false)






  }
}

//  定义样例类
case class Rating(useid:Int,movieid:Int,rating:Float,timestamp:Long)
